package com.atguigu.realtime.app.dwd;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_UserCartAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_UserCartAdd().init(
                3005,
                2,
                "Dwd_UserCartAdd"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tabEnv) {
        readOdsDb(tabEnv, "Dwd_UserCartAdd");
//        tabEnv.sqlQuery("select * from ods_db").execute().print();

        //从 ods_db中过滤出加购
        Table cartInfo = tabEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " `data`['course_id'] course_id, " +
                " ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='cart_info' " +
                "and `type`='insert' "
                );
//        tabEnv.createTemporaryView("cart_info", cartInfo);
//        tabEnv.sqlQuery("select * from cart_info").execute().print();

        //TODO 写出到 kafka中
        tabEnv.executeSql("create table dwd_user_cart_add(" +
                                "id string, " +
                                "user_id string, " +
                                "course_id string, " +
                                "ts bigint " +
                                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_USER_CART_ADD));

        cartInfo.executeInsert("dwd_user_cart_add");
    }
}
